package com.hujiang.basic.search.handler.impl;

import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.hujiang.basic.framework.plugin.mq.MqPlugin;
import com.hujiang.basic.search.api.IndexService;
import com.hujiang.basic.search.common.component.HJElasticSearchClient;
import com.hujiang.basic.search.common.converter.req.*;
import com.hujiang.basic.search.common.converter.resp.BulkConverter;
import com.hujiang.basic.search.model.*;
import com.hujiang.basic.search.zookeeper.zkLock.DistributedLock;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.index.reindex.BulkIndexByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


@Service
@Slf4j
public class IndexServiceImpl implements IndexService {

    @Autowired
    private HJElasticSearchClient client;

    @Autowired
    private IndexRequestConverter indexConverter;

    @Autowired
    private DistributedLock distributedLock;

    @Value(value = "${delay.es.queue}")
    private String delayEsQueue;

    @Value(value = "${zk.url}")
    private String zkAddr;

    @Value(value = "${increment.index.lock}")
    private String incLockPath;

    private static final Pattern pattern = Pattern.compile("\\d+$");

    private static final int OLD_INDICES_DAY = 7;
    private static final int UPDATE_IFCONFLICTED_SLEEPTIME = 500;

    @Override
    public HJIndexResponse add(HJIndexRequest req) {
        IndexRequest raw = indexConverter.convert(req);
        IndexResponse resp = client.index(raw);
        HJIndexResponse hjResp = new HJIndexResponse();
        hjResp.setUpdate(!resp.isCreated());
        BeanUtils.copyProperties(resp, hjResp);
        return hjResp;
    }


    @Override
    public HJDeleteResponse deleteOldIndices() {
        ImmutableOpenMap<String, IndexMetaData> oldIndices= client.getIndices();
        HJDeleteResponse hjResp = new HJDeleteResponse();
        for ( ObjectCursor<String> oldIndex : oldIndices.keys()) {
            Matcher matcher = pattern.matcher(oldIndex.value);
            if (matcher.find()) {
                int dateIndex = oldIndex.value.indexOf(matcher.group());
                if (oldIndices.get(oldIndex.value).getState() == IndexMetaData.State.CLOSE) {
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
                    try {
                        long date = simpleDateFormat.parse(oldIndex.value.substring(dateIndex)).getTime();
                        if (date <= (DateTime.now().minusDays(OLD_INDICES_DAY).getMillis())) {
                            client.deleteIndex(oldIndex.value);
                        }
                    } catch (Exception e) {
                        log.error("delete old indices error ",e);
                        hjResp.setDeleted(false);
                    }
                }
            }
        }
        hjResp.setDeleted(true);
        return hjResp;
    }



    @Autowired
    private DeleteRequestConverter deleteConverter;

    @Override
    public HJDeleteResponse delete(HJDeleteRequest req) {
        DeleteRequest raw = deleteConverter.convert(req);
        DeleteResponse resp = client.delete(raw);
        HJDeleteResponse hjResp = new HJDeleteResponse();
        BeanUtils.copyProperties(resp, hjResp);
        return hjResp;
    }

    @Autowired
    private UpdateRequestConverter updateConverter;

    @Override
    public HJUpdateResponse update(HJUpdateRequest req) {
        UpdateRequest raw = updateConverter.convert(req);
        UpdateResponse resp = client.update(raw);
        HJUpdateResponse hjResp = new HJUpdateResponse();
        BeanUtils.copyProperties(resp, hjResp);
        return hjResp;
    }

    @Autowired
    private BulkIndexRequestConverter bulkIndexConverter;

    @Autowired
    private BulkConverter bulkConverter;

    @Override
    public HJBulkResponse bulkAdd(HJBulkIndexRequest req) {
        BulkRequestBuilder raw = bulkIndexConverter.convert(req);
        BulkResponse resp = client.bulk(raw);
        return bulkConverter.convert(resp);
    }

    @Autowired
    private BulkDeleteRequestConverter bulkDeleteConverter;

    @Override
    public HJBulkResponse bulkDelete(HJBulkDeleteRequest req) {
        BulkRequestBuilder raw = bulkDeleteConverter.convert(req);
        BulkResponse resp = client.bulk(raw);
        return bulkConverter.convert(resp);
    }

    @Autowired
    private UpdateByQueryRequestConverter updateByQueryRequestConverter;


    @Override
    public HJUpdateByQueryResponse updateByQuery(HJUpdateByQueryRequest req) {
        UpdateByQueryRequest raw = updateByQueryRequestConverter.convert(req);
        raw.setRefresh(true);
        BulkIndexByScrollResponse resp;
        if (req.getConflicted() != null && req.getConflicted()) {
            log.info("receive conflicted message:{}", req);
            resp = updateByQueryIfConflicted(req);
        } else {
            resp = updateByQueryWhenNormal(req);
        }
        HJUpdateByQueryResponse hjResp = new HJUpdateByQueryResponse();
        if (resp != null) {
            BeanUtils.copyProperties(resp, hjResp);
            hjResp.setTookMillis(resp.getTook().millis());
        }
        return hjResp;
    }

    private BulkIndexByScrollResponse updateByQueryIfConflicted(HJUpdateByQueryRequest req) {
        try {
            distributedLock.incLock.acquire();
            Thread.sleep(UPDATE_IFCONFLICTED_SLEEPTIME);
            updateByQueryWhenNormal(req);
        } catch (Exception ex) {
            log.error("updateByQueryIfConflicted error ", ex);
        } finally {
            try {
                distributedLock.incLock.release();
            } catch (Exception e) {
                log.info("updateByQueryIfConflicted error ",e);
            }
        }
        return null;
    }


    private BulkIndexByScrollResponse updateByQueryWhenNormal(HJUpdateByQueryRequest req) {
        UpdateByQueryRequest raw = updateByQueryRequestConverter.convert(req);
        BulkIndexByScrollResponse resp = client.updateByQuery(raw);
        if (resp.getVersionConflicts() > 0) {
            putRequestToDelayEsMq(req);
        }
        return resp;
    }


    private void putRequestToDelayEsMq(HJUpdateByQueryRequest req) {
        req.setConflicted(true);
        MqPlugin.publisher().toQueue(delayEsQueue).build().send(req);
        log.info("put message to {} because conflict :{}", delayEsQueue, req);
    }


}
